#include <session.h>
#include <async_server.h>

Session::Session(boost::asio::io_context& ioct, AsyncServer* server) :
	socket_(ioct), server_(server), bClose_(false)
{
	sessionId_ = boost::uuids::to_string(boost::uuids::random_generator()());
	recvMsgHeader_ = std::make_shared<MsgNode>(MSG_HEADER_LEN);
}

void Session::startRead()
{
	this->asyncReadHeader();
}

void Session::asyncReadBody()
{
	this->asyncReadLen(recvMsgData_->totalLen_, 0, boost::bind(&Session::readBodyHandler, this, boost::placeholders::_1, boost::placeholders::_2));
}

bool Session::postSendNode(std::shared_ptr<SendNode> sendNode)
{
	std::lock_guard<std::mutex> locker(sendLock_);
	size_t queueSize = sendQueue_.size();
	if (!sendNode) {
		return false;
	}
	if (queueSize > MAX_QUEUE_SIZE) {
		std::cout << "session: " << getId() << " send queue fulled, size is " << queueSize << std::endl;
		return false;
	}
	this->sendQueue_.push(sendNode);
	if (queueSize > 0) { // 防止重复注册async_write事件，重复注册该事件可能导致发送包乱序
		return true;
	}
	this->asyncWrite();
	return true;
}

void Session::asyncWrite()
{
	auto topNode = sendQueue_.front();
	auto self = shared_from_this();
	socket_.async_write_some(boost::asio::buffer(topNode->data_, topNode->totalLen_),
		boost::bind(&Session::writeHandler, this, self, boost::placeholders::_1, boost::placeholders::_2));
}


void Session::asyncReadHeader()
{
	this->asyncReadLen(MSG_HEADER_LEN, 0, boost::bind(&Session::readHeaderHandler,this,boost::placeholders::_1,boost::placeholders::_2));
}

void Session::asyncReadLen(size_t len,size_t bytesHasRead, std::function<void(size_t bytesHasRead,const boost::system::error_code& ec)> handler)
{
	auto self = shared_from_this();
	socket_.async_read_some(boost::asio::buffer(buffer_ + bytesHasRead, len-bytesHasRead), [=](const boost::system::error_code& ec,size_t bytesRead) mutable {
		if (ec) {
			handler(bytesHasRead, ec);
			return;
		}
		bytesHasRead += bytesRead;
		if (bytesHasRead >= len) {
			handler(bytesHasRead, ec);
			return;
		}
		self->asyncReadLen(len, bytesHasRead, handler);
		}
	);
}

void Session::close()
{
	bClose_ = true;
	FileTransfferMgr::getInstance().removeFileStruct(sessionId_);
	socket_.close();
}

void Session::readHeaderHandler(size_t bytesRead, const boost::system::error_code& error)
{
	if (error) {
		std::cout << error.what() << std::endl;
		close();
		server_->clearSession(getId());
		return;
	}
	// note: 网络字节序转主机字节序
	recvMsgHeader_->clear();
	short msgId = 0;
	memcpy(&msgId, buffer_, MSG_ID_LEN);
	msgId = boost::asio::detail::socket_ops::network_to_host_short(msgId);
	unsigned long long msgLen = 0;
	memcpy(&msgLen, buffer_ + MSG_ID_LEN, MSG_DATA_LEN);
	msgLen = boost::asio::detail::socket_ops::host_to_network_long(msgLen);
	std::cout << "msgid: " << msgId << ", msg data len: " << msgLen << std::endl;
	if (msgLen > MSG_MAX_DATA_LEN) {
		std::cout << "invalid data length is " << msgLen << std::endl;
		server_->clearSession(getId());
		return;
	}
	// 创建接收数据节点，开始接收请求体
	recvMsgData_ = std::make_shared<RecvNode>(msgId, msgLen);
	this->asyncReadBody();
}

void Session::readBodyHandler(size_t bytesRead, const boost::system::error_code& error)
{
	if (error) {
		std::cout << error.what() << std::endl;
		close();
		server_->clearSession(getId());
		return;
	}
	if (bytesRead > bytesRead + MSG_HEADER_LEN) {
		std::cout << "bytesRead overflow: " << bytesRead << ", buffer maxSize:" << MSG_MAX_DATA_LEN - MSG_HEADER_LEN << std::endl;
		close();
		server_->clearSession(getId());
		return;
	}
	memcpy(recvMsgData_->data_, buffer_, recvMsgData_->totalLen_);
	std::cout << recvMsgData_->data_ << std::endl;
	// 后面把recvMsgData post到包处理线程，包处理线程处理后在post SendNode到这个线程
	std::shared_ptr<LogicNode> logicNode = std::make_shared<LogicNode>(shared_from_this(), recvMsgData_);
	if (recvMsgData_->msgId_ == static_cast<int>(MsgType::TYPE_TEXT)) {
		LogicSystem::getInstance().postTask(logicNode);
	}
	else if (recvMsgData_->msgId_ == static_cast<int>(MsgType::TYPE_FILE)) {
		FileTransfferSystem::getInstance().postTask(logicNode);
	}
	//std::shared_ptr<SendNode> sendNode = std::make_shared<SendNode>(recvMsgData_->data_, 1, recvMsgData_->totalLen_);
	//postSendNode(sendNode);
	this->asyncReadHeader();
}

void Session::writeHandler(std::shared_ptr<Session> self, const boost::system::error_code& ec, size_t bytesWrite)
{
	try {
		if (!ec) {
			std::lock_guard<std::mutex> locker(sendLock_);
			auto topNode = sendQueue_.front();
			topNode->curLen_ += bytesWrite;
			if (topNode->curLen_ < topNode->totalLen_) {
				socket_.async_write_some(boost::asio::buffer(topNode->data_ + topNode->curLen_, topNode->totalLen_ - topNode->curLen_),
					boost::bind(&Session::writeHandler, this, self, boost::placeholders::_1, boost::placeholders::_2));
				return;
			}
			if (topNode->curLen_ == topNode->totalLen_) {
				sendQueue_.pop();
			}
			if (sendQueue_.size() > 0) {
				this->asyncWrite();
			}
		}
		else {
			server_->clearSession(getId());
			close();
			std::cerr << "session:" << getId() << " handle async_write failed: " << ec.what() << std::endl;
		}
	}
	catch (const std::exception& error) {
		server_->clearSession(getId());
		close();
		std::cerr << error.what() << std::endl;
	}
}
